#ifndef MQCLIENT_MQ_CONNECTION_HPP_
#define MQCLIENT_MQ_CONNECTION_HPP_
#include "muduo/base/CountDownLatch.h"
#include "muduo/net/TcpClient.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "mq_channel.hpp"
#include "mq_worker.hpp"
#include <memory>
#include <string>

namespace mqclient 
{

using std::placeholders::_1;
using std::placeholders::_2;
using std::placeholders::_3;

class Connection
{
    using GoogleMessagePtr = std::shared_ptr<google::protobuf::Message>;
public:
    Connection(const std::string& svr_ip, uint16_t svr_port, const AsyncWorkerPtr& worker)
        : latch_(1)
        , client_(worker->loopthread_.startLoop(), muduo::net::InetAddress(svr_ip, svr_port), "Client")
        , dispatcher_(std::bind(&Connection::onUnknownMessage, this, _1, _2, _3))
        , codec_(std::make_shared<ProtobufCodec>(
            std::bind(&ProtobufDispatcher::onProtobufMessage, &dispatcher_, _1, _2, _3)))
        , worker_(worker)
        , channel_mg_(std::make_shared<ChannelManager>())
    {
        worker->pool_.start(); 
        dispatcher_.registerMessageCallback<proto::protocol::basicConsumeResponse>(
            std::bind(&Connection::consumeResponse, this, _1, _2, _3)
        );
        dispatcher_.registerMessageCallback<proto::protocol::basicCommonResponse>(
            std::bind(&Connection::basicResponse, this, _1, _2, _3)
        );

        client_.setConnectionCallback(std::bind(&Connection::onConnection, this, _1));
        client_.setMessageCallback(std::bind(&ProtobufCodec::onMessage, codec_.get(), _1, _2, _3));

        client_.connect();      // 非阻塞接口
        latch_.wait();          // 连接时进行阻塞，连接成功回调函数唤醒
    } 

    ChannelPtr openChannel()
    {
        ChannelPtr channel = channel_mg_->create(conn_, codec_);
        if(channel->openChannel() == false) {
            LOG_DEBUG << "open channel fail!\n";
            return ChannelPtr();
        }
        return channel;
    }
    void closeChannel(const ChannelPtr& channel)
    {
        channel->closeChannel();
        channel_mg_->remove(channel->channelId());
    }
private:
    void consumeResponse(const muduo::net::TcpConnectionPtr&,
                        const basicConsumeResponsePtr& message, muduo::Timestamp)
    {
        // 1. 找信道
        ChannelPtr channel = channel_mg_->getChannel(message->channel_id());
        if(channel.get() == nullptr) {
            LOG_DEBUG << "No channel found\n";
            return;
        } 
        // 2. 封装成异步任务, 提交到线程池
        worker_->pool_.submitTask([channel, message](){
            channel->consume(message);
        });
    }

    void basicResponse(const muduo::net::TcpConnectionPtr&,
                        const basicCommonResponsePtr& message, muduo::Timestamp)
    {
        // 1. 找信道
        ChannelPtr channel = channel_mg_->getChannel(message->channel_id());
        if(channel.get() == nullptr) {
            LOG_DEBUG << "No channel found.\n";
            return;
        } 
        // 2. 得到响应后, 反馈给信道
        channel->putBasicResponse(message);
    }

    void onConnection(const muduo::net::TcpConnectionPtr& conn)
    {
        if(conn->connected()) {
            conn_ = conn;
            latch_.countDown();          // 唤醒主线程连接时的阻塞
        } else {
            conn_.reset();               // 关闭连接
        }
    }

    void onUnknownMessage(const muduo::net::TcpConnectionPtr&,
                        const GoogleMessagePtr& message, muduo::Timestamp)
    {
        LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
    }
private:
    // muduo::CountDownLatch latch_;   
    // AsyncWorkerPtr worker_;                // 异步工作句柄
    // muduo::net::TcpClient client_;
    // muduo::net::TcpConnectionPtr conn_;
    // ProtobufCodecPtr codec_;
    // ProtobufDispatcher dispatcher_;
    // ChannelManagerPtr channel_mg_;         // 信道管理句柄

    muduo::CountDownLatch latch_;   
    AsyncWorkerPtr worker_;                // 异步工作句柄
    muduo::net::TcpConnectionPtr conn_;
    muduo::net::TcpClient client_;
    ProtobufDispatcher dispatcher_;
    ProtobufCodecPtr codec_;
    ChannelManagerPtr channel_mg_;         // 信道管理句柄
};

using ConnectionPtr = std::shared_ptr<Connection>;

}

#endif